package cn.le.listener;

import cn.le.util.WebSocketUtil;
import cn.le.vo.KafkaDownVO;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;

import java.io.IOException;
import java.util.List;

import static cn.le.util.Constant.DOWN_TOPIC;

@Component
public class DeviceListener {

    @KafkaListener(topics = DOWN_TOPIC)
    public void listener(String msg){
        System.out.println(msg);
        if (WebSocketUtil.isEmpty()){
            return;
        }
        KafkaDownVO vo = JSONObject.parseObject(msg,KafkaDownVO.class);

        List<WebSocketSession> socket = WebSocketUtil.findSocketConnect(String.valueOf(vo.getDeviceId()));
        if (CollectionUtils.isEmpty(socket)){
            return;
        }
        TextMessage textMessage = new TextMessage(msg);
        socket.forEach(session -> {
            try {
                session.sendMessage(textMessage);
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
    }
}
